package com.chukun.flink.stream.window.join;

import com.chukun.flink.stream.window.join.bean.ClickBean;
import com.chukun.flink.stream.window.join.bean.PrepareData;
import com.chukun.flink.stream.window.join.bean.UserTrade;
import com.chukun.flink.stream.window.join.extractor.ClickBeanEventTimestampExtractor;
import com.chukun.flink.stream.window.join.extractor.TradeEventTimestampExtractor;
import com.chukun.flink.stream.window.join.watermarker.ClickBeanWatermarkerGenerator;
import com.chukun.flink.stream.window.join.watermarker.TradeWatermarkerGenerator;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @author chukun
 * @version 1.0.0
 * @description CoGroupFunction基本操作
 * @createTime 2022年05月26日 23:49:00
 */
public class TumblingWindowCoGroupOperator {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<ClickBean> clickStream = env.fromCollection(PrepareData.getClicksData());

        DataStreamSource<UserTrade> tradeStream = env.fromCollection(PrepareData.getTradeData());

        // 设置原始用户点击数据的水印与事件时间
        SingleOutputStreamOperator<ClickBean> eventTimeClickStream = clickStream.assignTimestampsAndWatermarks(
                // 生成水印 watermarker
                WatermarkStrategy.forGenerator((ctx) -> new ClickBeanWatermarkerGenerator())
                        // 抽取事件产生的时间
                        .withTimestampAssigner((ctx) -> new ClickBeanEventTimestampExtractor())
        );

        // 设置原始用户交易数据的水印与事件时间
        SingleOutputStreamOperator<UserTrade> eventTimeTradeStream = tradeStream.assignTimestampsAndWatermarks(
                // 生成水印 watermarker
                WatermarkStrategy.forGenerator((ctx) -> new TradeWatermarkerGenerator())
                        // 抽取事件产生的时间
                        .withTimestampAssigner((ctx) -> new TradeEventTimestampExtractor())
        );

        CoGroupedStreams.WithWindow<ClickBean, UserTrade, String, TimeWindow> joinedWindow = eventTimeClickStream
                .coGroup(eventTimeTradeStream)
                // 根据eventTimeClickStream的用户名对数据流元素分组
                .where(ClickBean::getUser)
                // 根据eventTimeTradeStream的用户名对数据流元素分组
                .equalTo(UserTrade::getUserName)
                // 基于事件时间的滚动窗口，为1小时
                .window(TumblingEventTimeWindows.of(Time.hours(1)));

        // 在窗口中应用CoGroupFunction函数
        DataStream<String> applyCoGroupStream = joinedWindow.apply(new CoGroupFunction<ClickBean, UserTrade, String>() {

            @Override
            public void coGroup(Iterable<ClickBean> clickIt, Iterable<UserTrade> tradeIt, Collector<String> collector) throws Exception {
                Iterator<ClickBean> clickIterator = clickIt.iterator();
                while (clickIterator.hasNext()) {
                    ClickBean click = clickIterator.next();
                    Iterator<UserTrade> tradeIterator = tradeIt.iterator();
                    String message = "";
                    // 将eventTimeTradeStream数据流在该窗口中的元素的名称与终端拼接在message中
                    while (tradeIterator.hasNext()) {
                        UserTrade trade = tradeIterator.next();
                        message =message +"--"+trade.getUserName()+"--"+trade.getClient();
                    }
                    // 将eventTimeClickStream中的用户与url与message拼接输出
                    collector.collect(click.getUser() +"--->"+click.getUrl() +":"+message);
                }
            }
        });

        applyCoGroupStream.print("CoGroupFunction输出");

        env.execute("TumblingWindowCoGroupOperator");
    }
}
